package org.apache.pig.pigpen.util;

import java.io.File;
import java.io.FileInputStream;
import java.util.Map;
import java.util.Properties;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.pig.ExecType;
import org.apache.pig.PigServer;
import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
import org.apache.pig.impl.PigContext;
import org.apache.pig.pigpen.ConsoleLogger;
import org.apache.pig.pigpen.PigPenActivator;
import org.apache.pig.pigpen.PluginConfiguration;
import org.apache.pig.pigpen.preferences.PreferenceConstants;
import org.eclipse.jface.preference.IPreferenceStore;

public class PigServerPool {

	private static String JOB_TRACKER_LOCATION = "mapred.job.tracker";
	private static String FILE_SYSTEM_LOCATION = "fs.default.name";

	public static PigServer getPigServer() {
		IPreferenceStore store = PigPenActivator.getDefault().getPreferenceStore();
		String stringList = store.getString(PreferenceConstants.P_PROPERTIES);
		Map<String, String> PrefProperties = PluginConfiguration
				.parseString(stringList);
		String confPath = PrefProperties.get(PreferenceConstants.P_CONFPATH);
		ConsoleLogger logger = ConsoleLogger.getDefault();
		PigServer pigServer = null;

		if (confPath == null) {
			logger.clear();
			logger.setFocus();
			logger
					.error("Please set the ConfigurationPath variable in Pig Preferences (Window -> Preferences -> Pig) to point to the directory containing hadoop-site.xml and pig.properties");
			return pigServer;
		}
		logger.clear();
		if (confPath.charAt(confPath.length() - 1) != '/')
			confPath = confPath + "/";

		try {
			File pigPropertyFile = new File(confPath + "pig.properties");
			if (pigPropertyFile.exists()) {
				Properties sysProperties = new Properties();
				sysProperties.load(new FileInputStream(pigPropertyFile));
				for (Entry<Object, Object> e : sysProperties.entrySet()) {
					System.setProperty((String) e.getKey(), (String) e
							.getValue());
				}
			}
			Configuration conf = new Configuration();
			conf.addResource((new File(confPath + "core-site.xml")).toURI()
					.toURL());
			conf.addResource((new File(confPath + "hdfs-site.xml")).toURI()
					.toURL());
			conf.addResource((new File(confPath + "mapred-site.xml")).toURI()
					.toURL());
			Properties properties = ConfigurationUtil.toProperties(conf);
			
			pigServer = new PigServer(ExecType.MAPREDUCE, properties);
		} catch (Throwable e) {
			logger.setFocus();
			logger.error(e.getMessage());
			logger.error("Please set the ConfigurationPath variable in Pig Preferences (Window -> Preferences -> Pig) to point to the directory containing hadoop-site.xml and pig.properties");
		}
		PigContext pigContext = pigServer.getPigContext();
		logger.setFocus();
		logger.info("Connected to map-reduce job tracker at: "
				+ pigContext.getProperties().getProperty(JOB_TRACKER_LOCATION));
		logger.info("Connected to hadoop file system at: "
				+ pigContext.getProperties().getProperty(FILE_SYSTEM_LOCATION));
		return pigServer;
	}
}
